package edu.zju.gis.dldsj.controller;

import edu.zju.gis.dldsj.common.Result;
import edu.zju.gis.dldsj.config.CommonSetting;
import edu.zju.gis.dldsj.config.QueryConfig;
import edu.zju.gis.dldsj.constant.CodeConstants;
import edu.zju.gis.dldsj.constant.CodeType;
import edu.zju.gis.dldsj.constant.FunctionStatus;
import edu.zju.gis.dldsj.entity.*;
import edu.zju.gis.dldsj.entity.vo.TaskWithMonitor;
import edu.zju.gis.dldsj.model.Storage;
import edu.zju.gis.dldsj.service.*;
import edu.zju.gis.dldsj.tasks.MonitorTasks;
import edu.zju.gis.dldsj.tasks.TaskMonitor;
import edu.zju.gis.dldsj.utils.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.fs.Path;
import org.apache.tomcat.util.http.fileupload.servlet.ServletFileUpload;
import org.json.JSONArray;
import org.json.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
import org.springframework.web.multipart.MultipartHttpServletRequest;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.*;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.time.Instant;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 用于处理模型计算相关业务。包括模型注册和删除，任务提交、监控、历史查看，结果下载、预览等
 *
 * @author yanlo yanlong_lee@qq.com
 * @version 1.0 2018/08/01
 */
@Slf4j
@CrossOrigin
@Controller
@RequestMapping("/parallel")
public class ParallelModelController {
    @Autowired
    private CommonSetting setting;
    @Autowired
    private QueryConfig queryConfig;
    @Autowired
    private ParallelModelService parallelModelService;
    @Autowired
    private TaskRecordService taskRecordService;
    @Autowired
    private MonitorService monitorService;
    @Autowired
    private RoleModelService roleModelService;
    @Autowired
    private UserModelService userModelService;
    @Autowired
    private StorageService storageService;
    @Autowired
    private StorageFieldService storageFieldService;
    @Autowired
    private UserService userService;
    @Autowired
    private UserDataService userDataService;
    @Autowired
    private UserDataFieldService userDataFieldService;
    @Autowired
    private MonitorTasks monitorTasks;
    @Value("${spring.mvc.static-path-pattern}")
    private String staticPathPattern;

    @RequestMapping(value = "/register", method = RequestMethod.POST)
    @ResponseBody
    public String register(@SessionAttribute("userId") String userId, HttpServletRequest request) {
        Result<String> result = new Result<>();
        if (!ServletFileUpload.isMultipartContent(request)) {//判断是否为包含文件的表单数据
            result.setCode(CodeConstants.VALIDATE_ERROR)
                    .setMessage("必须有xml、jar、图片(png,jpg)三个文件被上传，或xml、py、zip（可选）、图片(png,jpg)四个文件被上传");
            return result.toString();
        }
        List<MultipartFile> files = ((MultipartHttpServletRequest) request).getFiles("file");//获取name为file的文件（表单中name可重复）
        MultipartFile jarFile = null;//executable file, jar or py
        MultipartFile xmlFile = null;
        MultipartFile picFile = null;
        MultipartFile zipFile = null;
        CodeType codeType = CodeType.JAVA;
        for (MultipartFile file : files) {
            String filename = file.getOriginalFilename();
            if (!file.isEmpty() && filename != null) {//不处理空文件
                String ext = filename.substring(filename.lastIndexOf(".")).toLowerCase();
                switch (ext) {
                    case ".xml":
                        xmlFile = file;
                        break;
                    case ".jar":
                        jarFile = file;
                        break;
                    case ".py":
                        jarFile = file;
                        codeType = CodeType.PYTHON;
                        break;
                    case ".png":
                    case ".jpg":
                        picFile = file;
                        break;
                    case ".zip":
                    case ".egg":
                        zipFile = file;
                        break;
                }
            }
        }
        if (jarFile == null || xmlFile == null) {
            result.setCode(CodeConstants.VALIDATE_ERROR);
            result.setMessage("xml与jar（或py）文件未上传或上传失败");
            return result.toString();
        }
        try {
            File tmpXml = new File(setting.getTemplatePath(), UUID.randomUUID().toString() + ".xml");
            xmlFile.transferTo(tmpXml);
            ParallelModelWithBLOBs model = XMLReader.readParallelModel(tmpXml);//读取并行模型模板文件
            if (model.getArtifactId() == null || model.getArtifactId().isEmpty()) {
                result.setCode(CodeConstants.VALIDATE_ERROR);
                result.setMessage("xml文件必须包含字段`artifactId`.");
                return result.toString();
            }
            File xmlPath = new File(setting.getTemplatePath(), model.getArtifactId() + ".xml");
            if (xmlPath.exists())
                FileUtil.deletePath(xmlPath);
            FileUtil.rename(tmpXml, xmlPath);//模板文件重命名为<模型名>.xml
            model.setXmlPath(xmlPath.getAbsolutePath());
            String ext = codeType == CodeType.PYTHON ? ".py" : ".jar";
            File jarPath = new File(setting.getJarPath(), model.getArtifactId() + ext);
            if (jarPath.exists())
                FileUtil.deletePath(jarPath);
            jarFile.transferTo(jarPath);//jar文件命名为<模型名>.jar
            model.setJarPath(jarPath.getAbsolutePath());
            if (picFile != null) {
                String picName = picFile.getOriginalFilename();
                picName = picName == null ? ".png" : picName.substring(picName.lastIndexOf("."));
                File picPath = new File(setting.getPicPath(), model.getArtifactId() + picName);
                if (picPath.exists())
                    FileUtil.deletePath(picPath);
                picFile.transferTo(picPath);//pic文件命名为<模型名>.<png/jpg>
                String picUrl = staticPathPattern.replace("*", "") + picPath.getName();
                model.setPicPath(picUrl);
            }
            if (zipFile != null) {
                String zipName = zipFile.getOriginalFilename();
                zipName = zipName == null ? ".zip" : zipName.substring(zipName.lastIndexOf("."));
                File zipPath = new File(setting.getJarPath(), model.getArtifactId() + zipName);
                if (zipPath.exists())
                    FileUtil.deletePath(zipPath);
                zipFile.transferTo(zipPath);//pic文件命名为<模型名>.<png/jpg>
                model.setJarPath(model.getJarPath() + "," + zipPath.getAbsolutePath());
            }

            String paths = model.getXmlPath() + "," + model.getJarPath() + "," + model.getPicPath();
            log.info("文件上传完成，文件保存路径为:" + paths);
            result.setBody(paths);
            parallelModelService.insert(model);//保存模型信息到数据库
            UserModel userModel = new UserModel();
            userModel.setId(UUID.randomUUID().toString());
            userModel.setModelId(model.getArtifactId());
            userModel.setUserId(userId);
            userModel.setRegistrationTime(Date.from(Instant.now()));
            userModel.setStatus(FunctionStatus.NORMAL.name());
            userModelService.upsert(userModel);//关联到当前用户
            result.setCode(CodeConstants.SUCCESS);
        } catch (Exception e) {
            log.error("模型注册失败", e);
            result.setCode(CodeConstants.SERVICE_ERROR).setMessage("模型注册失败：" + e.getMessage());
        }
        return result.toString();
    }

    @RequestMapping(value = "/unregister/{artifactId}", method = RequestMethod.DELETE)
    @ResponseBody
    public String unregister(@SessionAttribute("userId") String userId, @PathVariable String artifactId) {
        Result<String> result = new Result<>();
        try {
            ParallelModel model = parallelModelService.select(artifactId);
            parallelModelService.delete(artifactId);
            roleModelService.updateStatus(artifactId, FunctionStatus.DELETED.name());
            userModelService.delete(userId, artifactId);
            userModelService.updateStatus(artifactId, FunctionStatus.DELETED.name());
            //数据库中记录删除成功后，删除相关文件
            FileUtil.deletePath(model.getJarPath().split(","));
            FileUtil.deletePath(model.getXmlPath());
            if (Files.exists(Paths.get(model.getPicPath())))
                FileUtil.deletePath(model.getPicPath());
            result.setCode(CodeConstants.SUCCESS).setBody("success");
        } catch (Exception e) {
            log.error("模型解除注册失败", e);
            result.setCode(CodeConstants.DAO_ERROR).setMessage("模型解除注册失败：" + e.getMessage());
        }
        return result.toString();
    }

    @RequestMapping(value = "/get", method = RequestMethod.GET)
    @ResponseBody
    public String getAllModels(@SessionAttribute("roleId") String roleId, @SessionAttribute("userId") String userId) {
        Result<List<? extends ParallelModel>> result = new Result<>();
        try {
            Set<String> artifactIds = roleModelService.selectByRoleId(roleId).stream().map(RoleModel::getModelId).collect(Collectors.toSet());
            artifactIds.addAll(userModelService.get(userId).stream().map(UserModel::getModelId).collect(Collectors.toSet()));
            List<ParallelModel> models = parallelModelService.getByIdList(artifactIds);
            result.setCode(CodeConstants.SUCCESS).setBody(models);
        } catch (Exception e) {
            log.error("模型列表读取失败", e);
            result.setCode(CodeConstants.DAO_ERROR).setMessage("模型列表读取失败：" + e.getMessage());
        }
        return result.toString();
    }

    @RequestMapping(value = "/get/{artifactId}", method = RequestMethod.GET)
    @ResponseBody
    public String getModel(@PathVariable String artifactId) {
        Result<ParallelModelWithBLOBs> result = new Result<>();
        ParallelModelWithBLOBs model = parallelModelService.select(artifactId);
        if (model == null)
            result.setCode(CodeConstants.VALIDATE_ERROR).setMessage("找不到artifactId为`" + artifactId + "`的模型");
        else
            result.setCode(CodeConstants.SUCCESS).setBody(model);
        return result.toString();
    }

    @RequestMapping(value = "/use/{artifactId}", method = RequestMethod.POST)
    @ResponseBody
    public String postTask(@SessionAttribute("userName") String userName, @PathVariable String artifactId, @RequestBody String requestBody) {
        Result<String> result = new Result<>();
        ParallelModelWithBLOBs model = parallelModelService.select(artifactId);
        if (model == null) {
            result.setCode(CodeConstants.VALIDATE_ERROR).setMessage("找不到artifactId为`" + artifactId + "`的模型");
            return result.toString();
        } else {
            JSONObject inputs = new JSONObject(requestBody);
            String remarks = inputs.optString("customname");
            JSONArray params = inputs.getJSONArray("params");
            String jobName = model.getArtifactId() + System.currentTimeMillis();
            String cmd = parallelModelService.getCmd(model, jobName, params);
            if (cmd == null) {
                result.setCode(CodeConstants.SYSTEM_ERROR).setMessage("不支持的模型运行框架：" + model.getFrameworkType());
            } else
                try {
                    SSHHelper.runSSH(setting.getNameNode(), setting.getUsername(), setting.getPassword(), cmd, setting.getParallelFilePath());
                    log.info("a new application is submitted by " + userName + " with cmd[" + cmd + "]");
                    submitTask(jobName, userName, model, params, remarks, true);
                    result.setCode(CodeConstants.SUCCESS);
                    result.setBody(jobName);
                } catch (IOException e) {
                    log.error("任务提交失败", e);
                    result.setCode(CodeConstants.SYSTEM_ERROR).setMessage("任务提交失败" + e.getMessage());
                }
            return result.toString();
        }
    }

    @RequestMapping(value = "/kill/{jobName}", method = RequestMethod.POST)
    @ResponseBody
    public String killJob(@PathVariable String jobName) {
        Result<String> result = new Result<>();
        try {
            String applicationId = monitorService.getByName(jobName).getId();
            String cmd = setting.getHadoopHome() + "/bin/yarn application -kill " + applicationId;
            SSHHelper.runSSH(setting.getNameNode(), setting.getUsername(), setting.getPassword(), cmd, setting.getParallelFilePath());
            result.setCode(CodeConstants.SUCCESS).setMessage("任务中止成功");
        } catch (IOException e) {
            log.error("任务中止失败", e);
            result.setCode(CodeConstants.SERVICE_ERROR).setMessage("任务中止失败：" + e.getMessage());
        }
        return result.toString();
    }

    @RequestMapping(value = "/jobs/user", method = RequestMethod.GET)
    @ResponseBody
    public String getJobRecords(@SessionAttribute("userName") String userName, Boolean onlyFinished) {
        Result<List<TaskWithMonitor>> result = new Result<>();
        try {
            List<TaskRecord> tasks;
            if (onlyFinished != null && onlyFinished)
                tasks = taskRecordService.getByUserWithState(userName, Monitor.FinalStatus.SUCCEEDED);
            else
                tasks = taskRecordService.getByUser(userName);
            result.setCode(CodeConstants.SUCCESS);
            result.setBody(tasks.stream().map(task -> new TaskWithMonitor(task,
                    task.getState().equals(Monitor.State.FINISHED.name())
                            ? null : monitorService.getByName(task.getApplicationId()))).collect(Collectors.toList())
            );
        } catch (Exception e) {
            log.error("任务记录读取失败", e);
            result.setCode(CodeConstants.DAO_ERROR).setMessage("任务记录读取失败：" + e.getMessage());
        }
        return result.toString();
    }

    @RequestMapping(value = "/jobs/running", method = RequestMethod.GET)
    @ResponseBody
    public String getRunningJobs(@SessionAttribute("userName") String userName) {
        Result<List<TaskRecord>> result = new Result<>();
        List<TaskRecord> tasks = taskRecordService.getByUserWithState(userName, Monitor.FinalStatus.UNDEFINED);
        result.setCode(CodeConstants.SUCCESS);
        result.setBody(tasks);
        return result.toString();
    }

    @RequestMapping(value = "/jobs/{jobName}", method = RequestMethod.GET)
    @ResponseBody
    public String getRecord(@PathVariable String jobName) {
        Result<TaskRecord> result = new Result<>();
        if (jobName.startsWith("application_")) {
            Monitor monitor = monitorService.select(jobName);
            jobName = monitor.getName();
        }
        TaskRecord task = taskRecordService.select(jobName);
        if (task != null) {
            result.setCode(CodeConstants.SUCCESS).setBody(task);
        } else {
            result.setCode(CodeConstants.DAO_ERROR).setMessage("任务`" + jobName + "`不存在或已被删除");
        }
        return result.toString();
    }

    @RequestMapping(value = "/jobs/{jobName}/delete", method = RequestMethod.DELETE)
    @ResponseBody
    public String deleteRecord(@PathVariable String jobName) {
        Result<String> result = new Result<>();
        Monitor monitor;
        if (jobName.startsWith("application_")) {
            monitor = monitorService.select(jobName);
        } else {
            monitor = monitorService.getByName(jobName);
        }
        if (monitor == null) {
            result.setCode(CodeConstants.SERVICE_ERROR).setMessage("该记录不存在或已被删除");
        } else {
            if (monitor.getFinalStatus().equals(Monitor.FinalStatus.UNDEFINED.name())) {
                result.setCode(CodeConstants.VALIDATE_ERROR).setMessage("该任务未完成，请先停止该任务");
            } else {
                try {
                    monitorService.delete(monitor.getId());
                    //FileOperation.deletePath(new File(setting.getJobResultPath(), monitor.getName()));//删除结果目录，存在权限问题
                    TaskRecord record = taskRecordService.select(monitor.getName());
                    if (record != null) {
                        taskRecordService.delete(record.getApplicationId());
                        if (record.getToolId().startsWith("edu.zju.gis.dldsj.stat")) {
                            String resultAddr = record.getResultAddress();
                            if (!resultAddr.startsWith("file://"))
                                HDFSHelper.getInstance().deletePath(new Path(resultAddr), true);
                            String cmd;
                            if (setting.getUsername().equals("root"))
                                cmd = "rm -rf " + new File(setting.getJobResultPath(), monitor.getName()).getAbsolutePath();
                            else
                                cmd = "echo " + setting.getPassword() + " | sudo -S rm -rf " + new File(setting.getJobResultPath(), monitor.getName()).getAbsolutePath();
                            SSHHelper.runSSHWithOutput(setting.getNameNode(), setting.getUsername(), setting.getPassword(), cmd, setting.getParallelFilePath());
                        }
                    }
                    result.setCode(CodeConstants.SUCCESS).setBody("success");
                } catch (Exception e) {
                    log.error("任务记录删除失败", e);
                    result.setCode(CodeConstants.SERVICE_ERROR).setMessage("任务记录删除失败：" + e.getMessage());
                }
            }
        }

        return result.toString();
    }

    @RequestMapping(value = "/monitor/{jobName}", method = RequestMethod.GET)
    @ResponseBody
    public String getMonitor(@PathVariable String jobName) {
        Result<Monitor> result = new Result<>();
        Monitor monitor;
        if (jobName.startsWith("application_"))
            monitor = monitorService.select(jobName);
        else
            monitor = monitorService.getByName(jobName);
        if (monitor == null)
            result.setCode(CodeConstants.VALIDATE_ERROR).setMessage("该任务不存在或已被删除");
        else
            result.setCode(CodeConstants.SUCCESS).setBody(monitor);
        return result.toString();
    }

    @RequestMapping(value = "/result/download/{jobName}", method = RequestMethod.GET)
    public void downloadResult(@PathVariable String jobName, HttpServletResponse response) {
        if (jobName.startsWith("application_")) {
            Monitor monitor = monitorService.select(jobName);
            jobName = monitor.getName();
        }
        TaskRecord taskRecord = taskRecordService.select(jobName);
        String fullPath = taskRecord.getResultAddress();
        if (fullPath.contains("file://")) {
            fullPath = fullPath.substring("file://".length());
        } else {
            boolean flag = checkLocalResult(jobName);
            if (!flag) {
                try {
                    response.getOutputStream().println("任务`" + jobName + "`的结果不存在或已被删除");
                } catch (IOException e) {
                    log.error("系统错误", e);
                }
                return;
            }
            fullPath = setting.getJobResultPath() + "/" + jobName;
        }
        final String tarPath = fullPath + ".zip";
        File srcFile = new File(fullPath);
        File[] children = srcFile.listFiles();
        if (children == null || children.length == 0) {
            //空文件夹不会打包
            try {
                new File(fullPath, "空文件夹").createNewFile();
            } catch (IOException e) {
                log.error("文件创建失败", e);
                response.setStatus(500);
                try (PrintWriter writer = response.getWriter()) {
                    writer.println("文件下载失败");
                } catch (IOException ex) {
                    log.error("系统错误", ex);
                }
            }
        }

        ZipCompressorByAnt zip = new ZipCompressorByAnt(tarPath);
        zip.compressExe(fullPath, true);
        File file = new File(tarPath);
        try (InputStream fis = new BufferedInputStream(new FileInputStream(file)); OutputStream fos = response.getOutputStream()) {
            byte[] buffer = new byte[1024 * 1024 * 4];
            response.setContentType("application/octet-stream");
            //防止中文名乱码
            response.addHeader("Content-Disposition", "attachment;filename=" + new String(jobName.getBytes(), "ISO8859-1") + ".zip");
            response.addHeader("Content-Length", String.valueOf(file.length()));
            int length;
            while ((length = fis.read(buffer)) != -1)
                fos.write(buffer, 0, length);
            FileUtil.deletePath(tarPath);
        } catch (IOException e) {
            log.error("文件下载失败", e);
            response.setStatus(500);
            try (PrintWriter writer = response.getWriter()) {
                writer.println("文件下载失败：");
                writer.println(e.getMessage());
            } catch (IOException ex) {
                log.error("系统错误", ex);
            }
        }
    }

    @RequestMapping(value = "/result/preview/{jobName}", method = RequestMethod.GET)
    @ResponseBody
    public String previewResult(@PathVariable String jobName, Integer size, Integer ordinal) {
        Result<JSONArray> result = new Result<>();
        try {
            TaskRecord taskRecord = taskRecordService.select(jobName);
            if (taskRecord == null)
                result.setCode(CodeConstants.SERVICE_ERROR).setMessage("任务`" + jobName + "`不存在或已被删除");
            else {
                JSONArray jsons = getDataAsJSONArray(taskRecord, size, ordinal);
                result.setCode(CodeConstants.SUCCESS).setBody(jsons);
            }
        } catch (Exception e) {
            log.error("数据读取失败", e);
            result.setCode(CodeConstants.SERVICE_ERROR).setMessage("数据读取失败：" + e.getMessage());
        }
        return result.toString();
    }

    @RequestMapping(value = "/result/preview/{jobName}/geojson", method = RequestMethod.GET)
    @ResponseBody
    public String previewResultAsGeoJSON(@PathVariable String jobName, int size, Integer ordinal) {
        Result<JSONObject> result = new Result<>();
        try {
            TaskRecord taskRecord = taskRecordService.select(jobName);
            if (taskRecord == null)
                result.setCode(CodeConstants.SERVICE_ERROR).setMessage("任务`" + jobName + "`不存在或已被删除");
            else {
                JSONArray jsons = getDataAsJSONArray(taskRecord, size, ordinal);
                JSONObject geojson = GeometryUtil.toGeoJSON(jsons, jobName);
                result.setCode(CodeConstants.SUCCESS).setBody(geojson);
            }
        } catch (Exception e) {
            log.error("数据读取失败", e);
            result.setCode(CodeConstants.SERVICE_ERROR).setMessage("数据读取失败：" + e.getMessage());
        }
        return result.toString();
    }

    @RequestMapping(value = "/data/filter/{dataId}", method = RequestMethod.POST)
    @ResponseBody
    public String queryData(@SessionAttribute("userName") String userName, @PathVariable String dataId
            , @RequestBody String requestBody) {
        try {
            Storage storage = storageService.select(dataId);
            if (storage == null)
                return Result.error("没有ID为`" + dataId + "`的数据").toString();
            File jar = new File(setting.getJarPath(), "DataQuery.jar");
            if (!jar.exists())
                return Result.error("找不到jar文件：" + jar.getAbsolutePath()).toString();
            JSONObject requestParams = new JSONObject(requestBody);
            String fields = requestParams.optString("fields");
            if (fields.isEmpty()) {
                fields = storageFieldService.getByDataId(dataId).stream()
                        .sorted(Comparator.comparingInt(DataField::getId)).map(DataField::getFieldName)
                        .collect(Collectors.joining(","));
            }
            String timeRange = requestParams.optString("timeRange");
            String extent = requestParams.optString("extent");
            String spatialFilter = requestParams.optString("spatialFilter");
            String sql = requestParams.optString("sql");
            String tag = requestParams.optString("tag", storage.getMeta_name());
            String jobName = tag + "_" + System.currentTimeMillis();
            JSONArray params = new JSONArray(Arrays.asList(dataId, fields, timeRange, extent, spatialFilter, sql
                    , setting.getJobResultPath() + "/" + jobName));
            String cmd = setting.getSparkHome() +
                    "/bin/spark-submit" +
                    " --class " + queryConfig.getMainClass() +
                    " --master yarn --deploy-mode cluster --driver-memory " + queryConfig.getDriverMemory() +
                    " --num-executors " + queryConfig.getNumExecutors() +
                    " --executor-memory " + queryConfig.getExecutorMemory() +
                    " --executor-cores " + queryConfig.getExecutorCores() +
                    " --conf spark.default.parallelism=" + queryConfig.getParallelism() +
                    " --name " + jobName + " " +
                    setting.getJarPath() + "/" + queryConfig.getJar() + " " + params.join(" ");

            SSHHelper.runSSH(setting.getNameNode(), setting.getUsername(), setting.getPassword(), cmd, setting.getParallelFilePath());
            log.info("a new query application is submitted by " + userName + " with cmd[" + cmd + "]");
            ParallelModelWithBLOBs model = new ParallelModelWithBLOBs();
            model.setArtifactId(queryConfig.getMainClass());
            JSONArray outFields = new JSONArray();
            for (String field : fields.split(",")) {
                outFields.put(new JSONObject().put("name", field));
            }
            model.setOut(new JSONArray().put(new JSONObject().put("fields", outFields)).toString());
            model.setParameters(new JSONArray()
                    .put(new JSONObject().put("name", "dataId").put("in", 0).put("source", "platform"))
                    .put(new JSONObject().put("name", "fields"))
                    .put(new JSONObject().put("name", "timeRange"))
                    .put(new JSONObject().put("name", "extent"))
                    .put(new JSONObject().put("name", "spatialFilter"))
                    .put(new JSONObject().put("name", "sql"))
                    .put(new JSONObject().put("name", "output").put("out", 0))
                    .toString());
            submitTask(jobName, userName, model, params, tag, true);
            return Result.success().setBody(jobName).toString();
        } catch (Exception e) {
            log.error("任务提交失败", e);
            return Result.error().setMessage("任务提交失败" + e.getMessage()).toString();
        }
    }

    @RequestMapping(value = "/data/stat/{dataId}", method = RequestMethod.POST)
    @ResponseBody
    public String statData(@SessionAttribute("userName") String userName, @PathVariable String dataId
            , @RequestBody String requestBody) {
        try {
            Storage storage = storageService.select(dataId);
            if (storage == null)
                return Result.error("找不到ID为`" + dataId + "`的数据").toString();
            File jar = new File(setting.getJarPath(), queryConfig.getJar());
            if (!jar.exists())
                return Result.error("找不到jar文件：" + jar.getAbsolutePath()).toString();
            JSONObject requestParams = new JSONObject(requestBody);
            String fields = requestParams.optString("fields", "*");
            String timeRange = requestParams.optString("timeRange");
            String extent = requestParams.optString("extent");
            String spatialFilter = requestParams.optString("spatialFilter");
            String sql = requestParams.optString("sql");
            String group = requestParams.optString("groupField");
            if (group.isEmpty())
                group = "_";
            String stat = requestParams.getString("statField");
            String tag = requestParams.optString("tag", storage.getMeta_name());
            String jobName = tag + "_" + System.currentTimeMillis();
            String algo = requestParams.optString("algo", "basic");
            Map<String, String> algoConfig = queryConfig.getStatistics().getOrDefault(algo, null);
            if (algoConfig == null)
                return Result.error("the chosen algorithm is not supported.").toString();
            JSONArray params = new JSONArray(Arrays.asList(dataId, fields, timeRange, extent, spatialFilter, sql, group, stat
                    , setting.getJobResultPath() + "/" + jobName));
            String cmd = setting.getSparkHome() +
                    "/bin/spark-submit" +
                    " --class " + algoConfig.get("mainClass") +
                    " --master yarn --deploy-mode cluster --driver-memory " + algoConfig.getOrDefault("driverMemory", queryConfig.getDriverMemory()) +
                    " --num-executors " + algoConfig.getOrDefault("numExecutors", queryConfig.getNumExecutors() + "") +
                    " --executor-memory " + algoConfig.getOrDefault("executorMemory", queryConfig.getExecutorMemory()) +
                    " --executor-cores " + algoConfig.getOrDefault("executorCores", queryConfig.getExecutorCores() + "") +
                    " --conf spark.default.parallelism=" + algoConfig.getOrDefault("parallelism", queryConfig.getParallelism() + "") +
                    " --name " + jobName + " " +
                    setting.getJarPath() + "/" + this.queryConfig.getJar() + " " + params.join(" ");

            SSHHelper.runSSH(setting.getNameNode(), setting.getUsername(), setting.getPassword(), cmd, setting.getParallelFilePath());
            log.info("a new statistics application is submitted by " + userName + " with cmd[" + cmd + "]");
            ParallelModelWithBLOBs model = new ParallelModelWithBLOBs();
            model.setArtifactId(algoConfig.get("mainClass"));
            JSONArray outFields = new JSONArray().put(new JSONObject().put("name", group));
            for (String field : algoConfig.get("outFields").split(",")) {
                outFields.put(new JSONObject().put("name", field));
            }
            model.setOut(new JSONArray().put(new JSONObject().put("fields", outFields)).toString());
            model.setParameters(new JSONArray()
                    .put(new JSONObject().put("name", "dataId").put("in", 0).put("source", "platform"))
                    .put(new JSONObject().put("name", "fields"))
                    .put(new JSONObject().put("name", "timeRange"))
                    .put(new JSONObject().put("name", "extent"))
                    .put(new JSONObject().put("name", "spatialFilter"))
                    .put(new JSONObject().put("name", "sql"))
                    .put(new JSONObject().put("name", "group"))
                    .put(new JSONObject().put("name", "stat"))
                    .put(new JSONObject().put("name", "output").put("out", 0))
                    .toString());
            submitTask(jobName, userName, model, params, tag, true);
            return Result.success().setBody(jobName).toString();
        } catch (Exception e) {
            log.error("任务提交失败", e);
            return Result.error().setMessage("任务提交失败" + e.getMessage()).toString();
        }
    }

    private void submitTask(String jobName, String userName, ParallelModelWithBLOBs model, JSONArray params, String tag, boolean addToMyData) {
        TaskRecord record = new TaskRecord();
        record.setApplicationId(jobName);
        record.setToolId(model.getArtifactId());
        record.setCreateUser(userName);
        record.setSubmitTime(String.valueOf(System.currentTimeMillis()));
        record.setState(Monitor.FinalStatus.UNDEFINED.name());
        record.setRemarks(tag);
        List<String> fieldNames = new ArrayList<>();
        if (!model.getOut().trim().isEmpty()) {
            JSONArray outputs = new JSONArray(model.getOut());
            if (outputs.length() > 0) {
                JSONObject output = outputs.getJSONObject(0);
                JSONArray fields = output.optJSONArray("fields");
                for (int i = 0; i < fields.length(); i++) {
                    JSONObject field = fields.getJSONObject(i);
                    String name = field.optString("name");
                    if (!name.isEmpty()) {
                        fieldNames.add(name);
                    }
                }
            }
        }
        if (fieldNames.isEmpty())
            record.setOutputType("file");
        else
            record.setOutputType(String.join(",", fieldNames));
        record.setParams(params.join(","));
        taskRecordService.insert(record);
        monitorTasks.execute(new TaskMonitor(setting, taskRecordService, monitorService
                , userService, userDataService, userDataFieldService
                , record, model, params, addToMyData));
    }

    /**
     * 读取模型计算结果
     *
     * @param taskRecord 任务记录
     * @param size       读取记录的最大数量
     */
    private JSONArray getDataAsJSONArray(TaskRecord taskRecord, Integer size, Integer ordinal) throws IOException {
        final boolean skipFirstLine = taskRecord.getToolId().startsWith("edu.zju.gis.dldsj.stat");
        String[] resultAddresses = taskRecord.getResultAddress().split(",");
        if (ordinal == null)
            ordinal = 0;
        if (ordinal >= resultAddresses.length)
            ordinal = resultAddresses.length - 1;
        //1. 首先读取指定行数的结果
        List<String> lines = FileUtil.readByLine(resultAddresses[ordinal], size, skipFirstLine);
        //2. 再对结果进行列化处理
        JSONArray jsons = new JSONArray();
        String[] fields;
        if (taskRecord.getOutputType().equals("file"))
            fields = new String[]{"line"};
        else
            fields = taskRecord.getOutputType().split(",");
        if (fields.length < 2) {
            lines.forEach(line -> jsons.put(new JSONObject().put(fields.length == 1 ? fields[0] : "line", line)));
        } else {
            boolean isCsv = lines.stream().limit(20).noneMatch(s -> s.contains("\t"));
            lines.forEach(s -> {
                String[] values = s.split(isCsv ? "," : "\t", -1);
                JSONObject json = new JSONObject();
                for (int i = 0; i < fields.length; i++)
                    json.put(fields[i], values[i]);
                jsons.put(json);
            });
        }
        return jsons;
    }

    private List<String[]> getDataAsArray(TaskRecord taskRecord, Integer size) throws IOException {
        final boolean skipFirstLine = taskRecord.getToolId().startsWith("edu.zju.gis.dldsj.stat");
        String resultAddress = taskRecord.getResultAddress();
        //1. 首先读取指定行数的结果
        List<String> lines = FileUtil.readByLine(resultAddress, size, skipFirstLine);
        //2. 再对结果进行列化处理
        List<String[]> results = new ArrayList<>(size == null ? 10 : size);
        String[] fields;
        if (taskRecord.getOutputType().equals("file"))
            fields = new String[]{"line"};
        else
            fields = taskRecord.getOutputType().split(",");
        if (fields.length < 2) {
            results.add(new String[]{fields.length == 1 ? fields[0] : "line"});
            for (String line : lines)
                results.add(new String[]{line});
        } else {
            boolean isCsv = lines.stream().limit(20).noneMatch(s -> s.contains("\t"));
            results.add(fields);
            lines.forEach(s -> results.add(s.split(isCsv ? "," : "\t", -1)));
        }
        return results;
    }

    private boolean checkLocalResult(String jobName) throws IllegalArgumentException {
        TaskRecord record = taskRecordService.select(jobName);
        if (record == null)
            throw new IllegalArgumentException("任务`" + jobName + "`不存在或已被删除");
        File localFile = new File(setting.getJobResultPath(), jobName);
        if (!localFile.exists()) {
            String fullPath = record.getResultAddress();
            HDFSHelper hdfsHelper = HDFSHelper.getInstance();//copy from hdfs
            org.apache.hadoop.fs.Path src = new org.apache.hadoop.fs.Path(fullPath);
            try {
                if (hdfsHelper.exists(src)) {
                    org.apache.hadoop.fs.Path target = new org.apache.hadoop.fs.Path(localFile.getAbsolutePath());
                    return hdfsHelper.copyToLocalFile(src, target);
                } else
                    return false;
            } catch (Exception e) {
                log.error("文件`" + localFile.getAbsolutePath() + "`读取异常", e);
                return false;
            }
        } else
            return true;
    }
}
